package com.bw.fun;

import com.alibaba.fastjson.JSON;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/*
create table dim_base_dic (id varchar primary key , line varchar);
create table dim_sku_info (id varchar primary key , line varchar);
create table dim_spu_info (id varchar primary key , line varchar);
create table dim_user_info (id varchar primary key , line varchar);
create table dim_base_region (id varchar primary key , line varchar);
create table dim_coupon_info (id varchar primary key , line varchar);
create table dim_activity_sku (id varchar primary key , line varchar);
create table dim_coupon_range (id varchar primary key , line varchar);
create table dim_activity_info (id varchar primary key , line varchar);
create table dim_activity_rule (id varchar primary key , line varchar);
create table dim_base_province (id varchar primary key , line varchar);
create table dim_base_category1 (id varchar primary key , line varchar);
create table dim_base_category2 (id varchar primary key , line varchar);
create table dim_base_category3 (id varchar primary key , line varchar);
create table dim_base_trademark (id varchar primary key , line varchar);
create table dim_financial_sku_cost (id varchar primary key , line varchar);
 */
public class DimSinkFunc extends RichSinkFunction<String> {
    Connection connection = null;
    PreparedStatement ps = null;
    @Override
    public void open(Configuration parameters) throws Exception {

        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        connection = DriverManager.getConnection("jdbc:phoenix:hadoop-single:2181", "", "");
        connection.setAutoCommit(true);

    }

    @Override
    public void close() throws Exception {

        connection.close();
    }

    @Override
    public void invoke(String value, Context context) throws Exception {
        String tb = "dim_"+JSON.parseObject(value).getString("tb");
        String id = JSON.parseObject(value).getJSONObject("data").getString("id");
        ps = connection.prepareStatement("upsert into "+tb+" values (?,?)");
        ps.setString(1,id);
        ps.setString(2,value);
        ps.execute();
        ps.close();
    }
}
